#include "message_process_impl.h"

#include <core/QbLogger.h>
#include <core/sharedata/login_data.h>
#include <core/system/sleep.h>
#include <qbmessage/include/message_kernel.h>
#include <qbprotocol/include/rdMessage.h>
#include <string.h>

#include "../sync/message_sync_impl.h"

#ifndef WIN32
#include <arpa/inet.h>
#else
#pragma comment(lib, "ws2_32.lib")
#endif

namespace qb {
char CMessageProcessImpl::s_rcvBuf[Y_RD_SERVICE_MESSAGE_BLOCK_LEN] = {0};
char CMessageProcessImpl::s_preBuf[Y_RD_SERVICE_MESSAGE_BLOCK_LEN] = {0};

CMessageProcessImpl::CMessageProcessImpl()
    : m_event(*EventWrapper::Create(false)),
      m_thread(NULL),
      m_exit_flag(0),
      m_spNetworkService(NULL),
      m_spMessageSync(NULL) {}

CMessageProcessImpl::~CMessageProcessImpl() { delete &m_event; }

STDMETHODIMP_(int32_t)
CMessageProcessImpl::Create(THIS_ INetworkService* ins, IMessageSync* sync) {
  m_spNetworkService = ins;
  m_spMessageSync = sync;

  m_event.Reset();
  ++m_exit_flag;

  m_thread = ThreadWrapper::CreateThread(ThreadProcess, this, kNormalPriority,
                                         "message process thread");
  uint32_t threadid = 0;
  m_thread->Start(threadid);

  return 0;
}

STDMETHODIMP_(int32_t) CMessageProcessImpl::Destory(THIS_) {
  ////////////////////////////////////////////////////////////////////////////
  // remark1: 销毁动作必须在主线程中完成, 简化多线程管理逻辑
  // remark2: 这里认为模块的销毁时机一定是在所有调用此模块的子线程结束后
  ////////////////////////////////////////////////////////////////////////////

  // 清空观察者
  StopThread(&m_thread);

  m_spMessageSync = NULL;

  return 0;
}

STDMETHODIMP_(void)
CMessageProcessImpl::UnRegisterMsgProcessorOnThread(
    int msgType, IMessageProcessorOnSubThread* observer) {
  qb::base::ScopedLock<qb::base::Spinlock> locker(m_lock);
  auto iter = m_threadMsgProcessors.find(msgType);
  if (iter != m_threadMsgProcessors.end()) m_threadMsgProcessors.erase(iter);
}

STDMETHODIMP_(bool)
CMessageProcessImpl::RegisterMsgProcessorOnThread(
    int msgType, IMessageProcessorOnSubThread* observer) {
  qb::base::ScopedLock<qb::base::Spinlock> locker(m_lock);
  auto iter = m_threadMsgProcessors.find(msgType);
  if (iter != m_threadMsgProcessors.end()) {
    return false;
  }
  m_threadMsgProcessors[msgType] = observer;
  return true;
}

STDMETHODIMP_(int32_t)
CMessageProcessImpl::RegisterMessageProcessOnMainThreadObserver(
    THIS_ IMessageProcessOnMainThreadObserver* observer) {
  qb::base::ScopedLock<qb::base::Spinlock> locker(m_lock);
  int32_t nRet = RegisterObserver<IMessageProcessOnMainThreadObserver>(
      observer, m_observersMainThread);
  if (-1 == nRet) {
    log_warning("already enabled");
  }

  return nRet;
}

STDMETHODIMP_(int32_t)
CMessageProcessImpl::DeRegisterMessageProcessOnMainThreadObserver(
    THIS_ IMessageProcessOnMainThreadObserver* observer) {
  qb::base::ScopedLock<qb::base::Spinlock> locker(m_lock);
  int32_t nRet = DeRegisterObserver<IMessageProcessOnMainThreadObserver>(
      observer, m_observersMainThread);
  if (-1 == nRet) {
    log_warning("already enabled");
  }

  return nRet;
}

STDMETHODIMP_(int32_t)
CMessageProcessImpl::PushNetMessage(THIS_ MESSAGE_SHARED_PTR msp) {
  m_queue.push(msp);
  m_event.Set();

  return 0;
}

STDMETHODIMP_(int32_t)
CMessageProcessImpl::PushMainThreadMessage(THIS_ char* buff, int32_t buflen) {
  QB_CHECK_RETURN_INT1(buff && buflen > 0);

  xMsgHeadAck_t xHead;
  memcpy(&xHead, buff, sizeof(xHead));
  int msgtype = ntohl(xHead.msgtype);

  // 主线程回调中，默认在回调函数里处理完智能指针的内容，如果需要延长智能指针的生命周期，需要根据协议号，将makeSharedPtr的第三个参数置为true
  // BList<IMessageProcessOnMainThreadObserver*>::iterator iter =
  // m_observersMainThread.begin(); for_each_list(m_observersMainThread, iter,
  // (iter++)->first->OnMainThreadProcessData(msgtype, msp));
  return 0;
}

void CMessageProcessImpl::StopThread(ThreadWrapper** thread_) {
  while (m_exit_flag.Value() > 0) {
    --m_exit_flag;
  }
  m_event.Set();

  ThreadWrapper* tmpPtr = *thread_;
  if (tmpPtr) {
    tmpPtr->SetNotAlive();

    if (tmpPtr->Stop()) {
      delete tmpPtr;
    }
  }
}

bool CMessageProcessImpl::ThreadProcess(void* args) {
  return static_cast<CMessageProcessImpl*>(args)->ThreadProcessMessage(args);
}

bool CMessageProcessImpl::ThreadProcessMessage(void* args) {
  while (m_exit_flag.Value() > 0) {
    if (m_queue.empty()) {
      m_event.Wait(0xFFFFFFFF);
      if (m_exit_flag.Value() <= 0) {
        break;
      }

      if (m_queue.empty()) {
        continue;
      }
    }

    MESSAGE_SHARED_PTR msp = m_queue.pop();
    ProcessMessage(msp->getBuff(), msp->getLen());
    msp.reset();
  }

  return false;
}

bool CMessageProcessImpl::ProcessMessage(const char* buff, int buflen) {
  QB_CHECK_RETURN_BOOL1(buff && buflen > 0);
  QB_CHECK_RETURN_BOOL1(m_spMessageSync && m_spNetworkService);

  memset(s_rcvBuf, 0x00, Y_RD_SERVICE_MESSAGE_BLOCK_LEN);
  memset(s_preBuf, 0x00, Y_RD_SERVICE_MESSAGE_BLOCK_LEN);
  memcpy(s_rcvBuf, buff, buflen);

  int32_t preLen = 0;
  int retUnPack = rdMessage::iUnPack(s_rcvBuf, buflen, s_preBuf, &preLen);
  if (retUnPack < 0) {
    log_warning("Message Unpack Failed!");
    return false;
  }

  xMsgHeadAck_t xHead;
  memcpy(&xHead, s_preBuf, sizeof(xHead));
  xHead.msgtype = ntohl(xHead.msgtype);
  short retCode = ntohs(xHead.retcode);
  {
    qb::base::ScopedLock<qb::base::Spinlock> locker(m_lock);

    if (xHead.msgtype == E_FID_CS_FORCE_QUIT) {
      m_spNetworkService->NotifyForceQuit(loginData::instance().ConnectID());
      return true;
    }

    auto iter = m_threadMsgProcessors.find(xHead.msgtype);
    if (iter == m_threadMsgProcessors.end() || iter->second == nullptr) {
      return NotifyMainThreadObserver(xHead.msgtype, s_preBuf, preLen);
    }
    iter->second->OnResponse(xHead.msgtype, {s_preBuf, preLen, false, retCode});
  }
  return true;
}

bool CMessageProcessImpl::NotifyMainThreadObserver(int msgtype, char* buff,
                                                   int buflen) {
  (static_cast<CMessageSyncImpl*>(m_spMessageSync))->PushMessage(buff, buflen);
  return true;
}
}  // namespace qb
